/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package aiaudio.processing.base.mapreduce.standard.mappers;

import aiaudio.processing.base.mapreduce.CoordinateDoubleValueKey;
import aiaudio.database.tables.RatingsMatrixTable;
import java.io.IOException;
import java.util.Map;
import java.util.NavigableMap;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableMapper;

/**
 *
 * @author nastya
 */
public class UserToArtistValueMatrixMapper extends TableMapper<ImmutableBytesWritable, CoordinateDoubleValueKey> {
    private int numRecords = 0;

    @Override
    protected void map(ImmutableBytesWritable key, Result value, Context context) throws IOException, InterruptedException {
        try {
            NavigableMap<byte[], byte[]> familyMap = RatingsMatrixTable.getUsers(value);
            for (Map.Entry<byte[], byte[]> entry : familyMap.entrySet()) {
                String userName = RatingsMatrixTable.getTypedKey(entry);
                //TODO такое ощущение, что тут перепутаны user и artist
                if(RatingsMatrixTable.isDefaultValue(userName)){
                    continue;
                }
                double listenCount = RatingsMatrixTable.getTypedValue(entry);
                context.write(new ImmutableBytesWritable(entry.getKey()), new CoordinateDoubleValueKey(listenCount, key.get()));
            }
        } catch (InterruptedException e) {
            throw new IOException(e);
        }
        numRecords++;
        if ((numRecords % 1000) == 0) {
            context.setStatus("mapper processed " + numRecords + " records so far");
        }
    }
    
}
